package com.hmdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.StrUtil;
import com.hmdp.dto.Result;
import com.hmdp.entity.SeckillVoucher;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.hmdp.service.IVoucherOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.utils.RedisConstants;
import com.hmdp.utils.RedisIdWorker;
import com.hmdp.utils.UserHolder;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.aop.framework.AopContext;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * <p>
 * 服务实现类
 * </p>
 *
 * @author 虎哥
 * @since 2021-12-22
 */
@Slf4j
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {

    @Resource
    private ISeckillVoucherService seckillVoucherService;

    /*注入redis的id自动生成器*/
    @Resource
    private RedisIdWorker redisIdWorker;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private RedissonClient redissonClient;

    /*执行lua脚本*/
    private static final DefaultRedisScript<Long> SECKLL_SCRIPT;

    static {
        SECKLL_SCRIPT=new DefaultRedisScript<>();/*创建一个Redis的默认脚本对象*/
        SECKLL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
        SECKLL_SCRIPT.setResultType(Long.class);
    }

    /*根据优惠券id生成订单*/
    @Override
    public Result seckillVoucher(Long voucherId) {
        /*1.根据优惠券id去查询优惠券，判断优惠券是否可以使用*/
        SeckillVoucher seckillVoucher = seckillVoucherService.getById(voucherId);
        LocalDateTime nowTime = LocalDateTime.now();
        if(nowTime.isAfter(seckillVoucher.getEndTime()) || nowTime.isBefore(seckillVoucher.getBeginTime())){
            /*说明优惠券过期，不能抢购优惠券了*/
            return Result.fail("优惠券过期，不能抢购优惠券");
        }
        /*2.获取用户id，获取订单id*/
        Long userId = UserHolder.getUser().getId();
        if (StringUtils.isEmpty(userId)){
            return Result.fail("用户没有登录~");
        }
        long orderId = redisIdWorker.nextId("order");
        /*3.执行Lua脚本*/
        Long result = stringRedisTemplate.execute(SECKLL_SCRIPT, Collections.emptyList(), voucherId.toString(), userId.toString(), String.valueOf(orderId));
        /*4.得到lua脚本的返回值，判断其是否符合下单的条件*/
        if(result==1){
            return Result.fail("库存不足！");
        }
        if (result==2){
            return Result.fail("用户重复下单！");
        }
        /*订单信息已经放在了redis里面的stream的消息队列里面了*/
        /*5.如果符合下单的条件，这里直接返回订单的id就可以了*/
        return Result.ok(orderId);
    }

    /*开启一个线程池去处理stream类型消息队列里面的消息*/
    private static final ExecutorService SECKILL_ORDER_EXECUTOR=Executors.newSingleThreadExecutor();

    @PostConstruct/*类初始化完毕之后就立即执行这个方法*/
    private void init(){
        SECKILL_ORDER_EXECUTOR.submit(
            new VoucherOrderHandler()
        );
    }

    /*类初始化完毕之后，就去处理redis消息队列里面的消息*/
    private class VoucherOrderHandler implements Runnable{
        @Override
        public void run() {
            /*xreadgroup group g1 c1 count 1 block 2000 streams s1*/
            /*参数解释：g1表示消费者组名 c1消费者 count 1表示获取消息的条数 block 2000表示获取消息的时候，最长等待时间 streams s1表示队列名*/
            while (true){
                try {
                    /*1.读取消息*/
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create("stream.orders", ReadOffset.lastConsumed())
                    );
                    /*2.判断list里面是否有消息，如果没有消息，继续下一次循环*/
                    if(list.isEmpty() || list==null){
                        continue;
                    }
                    /*3.解析从队列里面获取到的消息，将其转换为VoucherOrder*/
                    MapRecord<String, Object, Object> entries = list.get(0);
                    Map<Object, Object> value = entries.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                    /*4.根据订单信息操作数据库：包括扣减库存，创建订单*/
                    /*  需要注意的是这里为了保证线程安全，需要设置分布式锁，根据用户id设置分布式锁，保证一人一单？*/
                    Long userId = voucherOrder.getUserId();
                    RLock redisLock = redissonClient.getLock(RedisConstants.LOCK_ORDER + userId);
                    boolean flag = redisLock.tryLock();
                    if(flag){
                        try {
                            /*如果获取到了锁，就可以去扣减库存、创建订单*/
                            createVoucherOrder(voucherOrder);
                        } catch (Exception e) {
                            e.printStackTrace();
                        } finally {
                            /*释放锁*/
                            redisLock.unlock();
                            /*确认消息,表示这个消息从队列里面消费完毕！*/
                            stringRedisTemplate.opsForStream().acknowledge("s1", "g1", entries.getId());
                        }
                    }else {
                        /*如果没有获取到锁，说明原先这个用户已经下过一次单了*/
                        log.error("不允许重复下单~");
                        return;
                    }
                } catch (Exception e) {
                    log.error("获取redis消息队列里面的订单异常:" + e);
                    handlePendingList();
                    e.printStackTrace();
                }
            }
        }
    }

    /*
    * XADD stream.orders * field1 value1 field2 value2
    * XGROUP CREATE stream.orders g1 $ MKSTREAM
    * XACK stream.orders g1 message-id
    * XREADGROUP GROUP g1 c1 COUNT 1 STREAMS stream.orders >
    *
    * */
    public void handlePendingList(){
        /*xreadgroup group g1 c1 count 1 block 2000 streams s1*/
        /*参数解释：g1表示消费者组名 c1消费者 count 1表示获取消息的条数 block 2000表示获取消息的时候，最长等待时间 streams s1表示队列名*/
        while (true){
            try {
                /*1.读取消息*/
                List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                        Consumer.from("g1", "c1"),
                        StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                        StreamOffset.create("stream.orders", ReadOffset.lastConsumed())
                );
                /*2.判断list里面是否有消息，如果没有消息，继续下一次循环*/
                if(list.isEmpty() || list==null){
                    continue;
                }
                /*3.解析从队列里面获取到的消息，将其转换为VoucherOrder*/
                MapRecord<String, Object, Object> entries = list.get(0);
                Map<Object, Object> value = entries.getValue();
                VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                /*4.根据订单信息操作数据库：包括扣减库存，创建订单*/
                /*  需要注意的是这里为了保证线程安全，需要设置分布式锁，根据用户id设置分布式锁，保证一人一单？*/
                Long userId = voucherOrder.getUserId();
                RLock redisLock = redissonClient.getLock(RedisConstants.LOCK_ORDER + userId);
                boolean flag = redisLock.tryLock();
                if(flag){
                    try {
                        /*如果获取到了锁，就可以去扣减库存、创建订单*/
                        createVoucherOrder(voucherOrder);
                    } catch (Exception e) {
                        e.printStackTrace();
                    } finally {
                        /*释放锁*/
                        redisLock.unlock();
                        /*确认消息,表示这个消息从队列里面消费完毕！*/
                        stringRedisTemplate.opsForStream().acknowledge("s1", "g1", entries.getId());
                    }
                }else {
                    /*如果没有获取到锁，说明原先这个用户已经下过一次单了*/
                    log.error("不允许重复下单~");
                    return;
                }
            } catch (Exception e) {
                try {
                    Thread.sleep(20);
                } catch (InterruptedException interruptedException) {
                    interruptedException.printStackTrace();
                }
                e.printStackTrace();
            }
        }
    }

    /*创建订单*/
    @Transactional
    public void createVoucherOrder(VoucherOrder voucherOrder) {
        /*前面的是在redis里面进行模拟的，现在才是在数据库里面进行操作*/
        /*1.根据用户id判断是不是一人一单---redis的lua脚本里面已经判断过了 */
        /*2.判断库存是否充足---redis的lua脚本里面已经判断过了*/
        /*扣减库存(乐观锁-防止超卖)，进行下单*/
        boolean success = seckillVoucherService.update().setSql("stock=stock-1")
                .eq("voucher_id", voucherOrder.getVoucherId().toString())
                .gt("stock", 0).update();
        if (!success) {
            log.error("库存不足！");
        }
        /*下单*/
        baseMapper.insert(voucherOrder);
    }

}
